package com.zhang.gmall.app.dwd;

import com.zhang.gmall.utils.KafkaUtil;
import com.zhang.gmall.utils.MyEnv;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @title: 订单退款宽表
 * @author: zhang
 * @date: 2022/3/25 22:52
 */
public class OrderRefundWideApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取环境
        StreamExecutionEnvironment env = MyEnv.getStreamingEnv(4, false);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //join时候状态保留时间
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));

        //TODO 2.读取kafka topic_db主题数据
        String sourceTopic = "topic_db";
        String groupId = "order_refund_wide_app_group";
        String sinkTopic = " dwd_order_refund_wide";
        tableEnv.executeSql("" +
                "create table ods_db( " +
                "   `database` string," +
                "   `table` string," +
                "   `type` string," +
                "   `data` map<string,string>" +
                ")" + KafkaUtil.getKafkaDDL(sourceTopic, groupId));

        //TODO 3.读取kafka dwd_order_wide 主题数据
        tableEnv.executeSql("" +
                "create table order_wide( " +
                "    order_detail_id String, " +
                "    order_id String, " +
                "    sku_id String, " +
                "    sku_name String, " +
                "    order_price String, " +
                "    sku_num String, " +
                "    create_time String, " +
                "    source_type String, " +
                "    source_id String, " +
                "    split_total_amount String, " +
                "    split_activity_amount String, " +
                "    split_coupon_amount String, " +
                "    consignee String, " +
                "    consignee_tel String, " +
                "    total_amount String, " +
                "    order_status String, " +
                "    user_id String, " +
                "    payment_way String, " +
                "    delivery_address String, " +
                "    order_comment String, " +
                "    out_trade_no String, " +
                "    trade_body String, " +
                "    operate_time String, " +
                "    expire_time String, " +
                "    process_status String, " +
                "    tracking_no String, " +
                "    parent_order_id String, " +
                "    province_id String, " +
                "    activity_reduce_amount String, " +
                "    coupon_reduce_amount String, " +
                "    original_total_amount String, " +
                "    feight_fee String, " +
                "    feight_fee_reduce String, " +
                "    refundable_time String, " +
                "    order_activity_id String, " +
                "    activity_id String, " +
                "    activity_rule_id String, " +
                "    order_coupon_id String, " +
                "    coupon_id String, " +
                "    coupon_use_id String " +
                ")" + KafkaUtil.getKafkaDDL("dwd_order_wide", groupId));
        //tableEnv.sqlQuery("select * from order_wide").execute().print();

        //TODO 4.过滤退单数据表数据
        Table orderRefundTable = tableEnv.sqlQuery("" +
                "select " +
                "       data['id'] id," +
                "       data['user_id'] user_id," +
                "       data['order_id'] order_id," +
                "       data['sku_id'] sku_id," +
                "       data['refund_type'] refund_type," +
                "       data['refund_num'] refund_num," +
                "       data['refund_amount'] refund_amount," +
                "       data['refund_reason_type'] refund_reason_type," +
                "       data['refund_reason_txt'] refund_reason_txt," +
                "       data['refund_status'] refund_status," +
                "       data['create_time'] create_time " +
                "from ods_db " +
                "where  `database` = 'gmall_flink_2022' " +
                "and `table` = 'order_refund_info' " +
                "and `type` = 'insert'");
        tableEnv.createTemporaryView("refund_info", orderRefundTable);
        //tableEnv.sqlQuery("select * from refund_info").execute().print();

        //TODO 5.join退单表和订单宽表
        Table result = tableEnv.sqlQuery("" +
                "select " +
                "    ow.order_detail_id, " +
                "    ow.order_id, " +
                "    ow.sku_id, " +
                "    ow.sku_name, " +
                "    ow.order_price, " +
                "    ow.sku_num, " +
                "    ow.create_time, " +
                "    ow.source_type, " +
                "    ow.source_id, " +
                "    ow.split_total_amount, " +
                "    ow.split_activity_amount, " +
                "    ow.split_coupon_amount, " +
                "    ow.consignee, " +
                "    ow.consignee_tel, " +
                "    ow.total_amount, " +
                "    ow.order_status, " +
                "    ow.user_id, " +
                "    ow.payment_way, " +
                "    ow.delivery_address, " +
                "    ow.order_comment, " +
                "    ow.out_trade_no, " +
                "    ow.trade_body, " +
                "    ow.operate_time, " +
                "    ow.expire_time, " +
                "    ow.process_status, " +
                "    ow.tracking_no, " +
                "    ow.parent_order_id, " +
                "    ow.province_id, " +
                "    ow.activity_reduce_amount, " +
                "    ow.coupon_reduce_amount, " +
                "    ow.original_total_amount, " +
                "    ow.feight_fee, " +
                "    ow.feight_fee_reduce, " +
                "    ow.refundable_time, " +
                "    ow.order_activity_id, " +
                "    ow.activity_id, " +
                "    ow.activity_rule_id, " +
                "    ow.order_coupon_id, " +
                "    ow.coupon_id, " +
                "    ow.coupon_use_id, " +
                "    ri.id order_refund_id, " +
                "    ri.refund_type, " +
                "    ri.refund_num, " +
                "    ri.refund_amount, " +
                "    ri.refund_reason_type, " +
                "    ri.refund_reason_txt, " +
                "    ri.refund_status, " +
                "    ri.create_time order_refund_create_time " +
                "from refund_info ri " +
                "join order_wide ow " +
                "on ri.order_id = ow.order_id " +
                "and ri.sku_id = ow.sku_id ");
        tableEnv.createTemporaryView("result_table", result);
        tableEnv.toChangelogStream(result).print();


        //TODO 6.创建sink表将数据写入kafka
        tableEnv.executeSql("" +
                "create table order_refund_wide( " +
                "    order_detail_id string, " +
                "    order_id string, " +
                "    sku_id string, " +
                "    sku_name string, " +
                "    order_price string, " +
                "    sku_num string, " +
                "    create_time string, " +
                "    source_type string, " +
                "    source_id string, " +
                "    split_total_amount string, " +
                "    split_activity_amount string, " +
                "    split_coupon_amount string, " +
                "    consignee string, " +
                "    consignee_tel string, " +
                "    total_amount string, " +
                "    order_status string, " +
                "    user_id string, " +
                "    payment_way string, " +
                "    delivery_address string, " +
                "    order_comment string, " +
                "    out_trade_no string, " +
                "    trade_body string, " +
                "    operate_time string, " +
                "    expire_time string, " +
                "    process_status string, " +
                "    tracking_no string, " +
                "    parent_order_id string, " +
                "    province_id string, " +
                "    activity_reduce_amount string, " +
                "    coupon_reduce_amount string, " +
                "    original_total_amount string, " +
                "    feight_fee string, " +
                "    feight_fee_reduce string, " +
                "    refundable_time string, " +
                "    order_activity_id string, " +
                "    activity_id string, " +
                "    activity_rule_id string, " +
                "    order_coupon_id string, " +
                "    coupon_id string, " +
                "    coupon_use_id string, " +
                "    order_refund_id string, " +
                "    refund_type string, " +
                "    refund_num string, " +
                "    refund_amount string, " +
                "    refund_reason_type string, " +
                "    refund_reason_txt string, " +
                "    refund_status string, " +
                "    order_refund_create_time string, " +
                "    PRIMARY KEY (order_refund_id) NOT ENFORCED " +
                ")" + KafkaUtil.getUpsertKafkaDDL(sinkTopic));
        tableEnv.executeSql("insert into order_refund_wide select * from result_table");

        //TODO 7.启动任务
        env.execute();

    }
}
